@*
* Copyright 2015 Yahoo Inc. Licensed under the Apache License, Version 2.0
* See accompanying LICENSE file.
*@
@import kafka.manager.model.ActorModel.ConsumerType
@import models.form.ReassignPartitionOperation
@import models.form.ReassignPartitionOperation.ForceRunAssignment
@import kafka.manager.utils.LongFormatted
@(cluster:String
, topic: String
, topicIdentity: kafka.manager.model.ActorModel.TopicIdentity
, consumerList: Iterable[(String, ConsumerType)]
, reassignPartitionOperation: ReassignPartitionOperation
)(implicit af: features.ApplicationFeatures, messages: play.api.i18n.Messages, request:RequestHeader)

@getUnderReplicatedLevel(percentage: Int) = {
    @percentage match {
        case i if i > 50 => {table-danger}
        case i if i > 0 => {table-warning}
        case i => {}
    }
}

@getPreferredReplicasLevel(percentage: Int) = {
    @percentage match {
        case i if i > 50 && i<=75 => {table-warning}
        case i if i <=  50 => {table-danger}
        case i => {}
    }
}

@getBrokersSkewedLevel(percentage: Int) = {
    @percentage match {
        case i if i > 0 && i <= 33 => {table-warning}
        case i if i >= 34 => {table-danger}
        case i => {}
    }
}

@getBrokersLeaderSkewedLevel(percentage: Int) = {
    @percentage match {
        case i if i > 0 && i <= 33 => {table-warning}
        case i if i >= 34 => {table-danger}
        case i => {}
    }
}

@getBrokersSpreadLevel(percentage: Int) = {
    @percentage match {
        case i if i > 50 && i<=75 => {table-warning}
        case i if i <=  50 => {table-danger}
        case i => {}
    }
}

@getPartitionLeaderLevel(leader: Int) = {
    @leader match {
        case i if i < 0 => {table-danger}
        case i => {}
    }
}

@getIsUnderReplicatedLevel(underReplicated: Boolean) = {
    @underReplicated match {
        case true => {table-warning}
        case false => {}
    }
}

@getIsPreferredLeaderLevel(preferredReplica: Boolean) = {
    @preferredReplica match {
        case false => {table-warning}
        case true => {}
    }
}

@getBrokerIsSkewedLevel(isSkewed: Boolean) = {
    @isSkewed match {
        case true => {table-warning}
        case false => {}
    }
}

@renderTopicMetrics = {
    @if(topicIdentity.clusterContext.clusterFeatures.features(kafka.manager.features.KMJMXMetricsFeature)) {
        @views.html.common.brokerMetrics(topicIdentity.metrics)
    } else {
        <div class="alert alert-info" role="alert">
            Please enable JMX polling <a href="@routes.Cluster.updateCluster(cluster)" class="alert-link">here</a>.
        </div>
    }
}

@renderConsumerList = {
    @if(topicIdentity.clusterContext.config.pollConsumers){
        <table class="table">
            <tbody>
            @for((c:String, ct: ConsumerType) <- consumerList) {
                <tr>
                    <td><a href="@routes.Consumer.consumerAndTopic(cluster,c,topic, ct.toString)">@c</a></td>
                    <td>@ct.toString</td>
                </tr>
            }
            </tbody>
        </table>
    } else {
        <div class="alert alert-info" role="alert">
            Please enable consumer polling <a href="@routes.Cluster.updateCluster(cluster)" class="alert-link">here</a>.
        </div>
    }
}


<div class="row">
    <div class="col-md-5">
        <div class="card">
            <div class="card-header"><h4>Topic Summary</h4></div>
            <div class="card-body">
            <table class="table">
                <tbody>
                <tr>
                    <td>Replication</td>
                    <td>@topicIdentity.replicationFactor</td>
                <tr>
                    <td>Number of Partitions</td>
                    <td>@topicIdentity.partitions</td>
                </tr>
                <tr>
                    <td><span title="Represents the theoretical size of the topic">Sum of partition offsets</span></td>
                    <td>@topicIdentity.summedTopicOffsets.formattedAsDecimal</td>
                </tr>
                <tr>
                    <td>Total number of Brokers</td>
                    <td>@topicIdentity.numBrokers</td>
                </tr>
                <tr>
                    <td>Number of Brokers for Topic</td>
                    <td>@topicIdentity.topicBrokers</td>
                </tr>
                <tr>
                    <td><span title="Percentage of replicas leader being preferred (first in the list of replicas)">Preferred Replicas %</span></td>
                    <td class="@getPreferredReplicasLevel(topicIdentity.preferredReplicasPercentage)">
                        @topicIdentity.preferredReplicasPercentage
                    </td>
                </tr>
                <tr>
                    <td><span title="Percentage of brokers having more partitions than the average">Brokers Skewed %</span></td>
                    <td class="@getBrokersSkewedLevel(topicIdentity.brokersSkewPercentage)">
                        @topicIdentity.brokersSkewPercentage
                    </td>
                </tr>
                <tr>
                    <td><span title="Percentage of brokers having more partitions as leader than the average">Brokers Leader Skewed %</span></td>
                    <td class="@getBrokersLeaderSkewedLevel(topicIdentity.brokersLeaderSkewPercentage)">
                        @topicIdentity.brokersLeaderSkewPercentage
                    </td>
                </tr>
                <tr>
                    <td><span title="Percentage of cluster brokers having partitions from the topic">Brokers Spread %</span></td>
                    <td class="@getBrokersSpreadLevel(topicIdentity.brokersSpreadPercentage)">
                        @topicIdentity.brokersSpreadPercentage
                    </td>
                </tr>
                <tr>
                    <td><span title="Percentage of partitions having a missing replica">Under-replicated %</span></td>
                    <td class="@getUnderReplicatedLevel(topicIdentity.underReplicatedPercentage)">
                        @topicIdentity.underReplicatedPercentage
                    </td>
                </tr>
                @if(topicIdentity.clusterContext.config.jmxEnabled && topicIdentity.clusterContext.config.displaySizeEnabled){
                <tr>
                    <td>Leader Size</td>
                    <td>@topicIdentity.size</td>
                </tr>
                }
                </tbody>
            </table>
            @if(!topicIdentity.config.isEmpty) {
            <table class="table">
                <thead>
                <th>Config</th><th>Value</th>
                </thead>
                <tbody>
                @for( (k,v) <- topicIdentity.config) {
                <tr>
                    <td>@k</td>
                    <td>@v</td>
                </tr>
                }
                </tbody>
            </table>
            }
            </div>
        </div>
        <div class="card">
            <div class="card-header"><h4>Metrics</h4></div>
            <div class="card-body">
            @renderTopicMetrics
            </div>
        </div>
    </div>
    <div class="col-md-7">
        @features.app(features.KMTopicManagerFeature) {
        <div class="card">
            <div class="card-header"><h4>Operations</h4></div>
            <div class="card-body">
            <table class="table">
                <tbody>
                <tr>
                    @if(topicIdentity.clusterContext.clusterFeatures.features(kafka.manager.features.KMDeleteTopicFeature)) {
                    <td>
                        <a href="@routes.Topic.confirmDeleteTopic(cluster,topic)" class='btn btn-primary btn-block'>Delete Topic</a>
                    </td>
                    }
                    @features.app(features.KMReassignPartitionsFeature) {
                    <td>
                        @b4.vertical.form(routes.ReassignPartitions.handleOperation(cluster,topic)) { implicit fc =>
                            @reassignPartitionOperation match {
                                case ForceRunAssignment => {
                                    <fieldset>
                                        <input type="hidden" name="operation" value="force">
                                        <input type="hidden" name="brokers" value="@Seq()">
                                        @b4.submit('class -> "btn btn-primary btn-block"){ Force Reassign Partitions }
                                    </fieldset>
                                }
                                case _ => {
                                    <fieldset>
                                        <input type="hidden" name="operation" value="run">
                                        <input type="hidden" name="brokers" value="@Seq()">
                                        @b4.submit('class -> "btn btn-primary btn-block"){ Reassign Partitions }
                                    </fieldset>
                                }
                            }
                        }
                    </td>
                    <td>
                        <a href="@routes.ReassignPartitions.confirmAssignment(cluster,topic)" class="submit-button btn btn-primary" role="button">Generate Partition Assignments</a>
                    </td>
                    }
                </tr>
                <tr>
                    <td>
                        <a href="@routes.Topic.addPartitions(cluster,topic)" class="submit-button btn btn-primary" role="button">Add Partitions</a>
                    </td>
                    <td>
                        <a href="@routes.Topic.updateConfig(cluster,topic)" class="submit-button btn btn-primary" role="button">Update Config</a>
                    </td>
                    @features.app(features.KMReassignPartitionsFeature) {
                    <td>
                        <a href="@routes.ReassignPartitions.manualAssignments(cluster, topic)" class="submit-button btn btn-primary" role="button">Manual Partition Assignments</a>
                    </td>
                    }
                </tr>
                </tbody>
            </table>
            </div>
        </div>
        }
        <div class="card">
            <div class="card-header"><h4>Partitions by Broker</h4></div>
            <div class="card-body">
            <table class="table" style="table-layout: fixed; width: 100%">
                <thead>
                <tr><th>Broker</th><th># of Partitions</th><th># as Leader</th><th>Partitions</th><th><span title="Broker has more partitions than the average">Skewed?</span></th><th><span title="Broker has more partitions as leaders than the average">Leader Skewed?</span></th></tr>
                </thead>
                <tbody>
                @for(btp <- topicIdentity.partitionsByBroker) {
                <tr>
                    <td><a href="@routes.Cluster.broker(cluster,btp.id)">@btp.id</a></td>
                    <td>@btp.partitions.size</td>
                    <td>@btp.leaders.size</td>
                    <td style="word-wrap: break-word">@btp.partitions.mkString("(",",",")")</td>
                    <td class="@getBrokerIsSkewedLevel(btp.isSkewed)">
                        @btp.isSkewed
                    </td>
                    <td class="@getBrokerIsSkewedLevel(btp.isLeaderSkewed)">
                        @btp.isLeaderSkewed
                    </td>
                </tr>
                }
                </tbody>
            </table>
            </div>
        </div>
        <div class="card">
            <div class="card-header"><h4>Consumers consuming from this topic</h4></div>
            <div class="card-body">
            @renderConsumerList
            </div>
        </div>
    </div>
</div>
<div class="row">
    <div class="col-md-12">
        <div class="card">
            <div class="card-header"><h4>Partition Information</h4></div>
            <div class="card-body">
            <table class="table">
                <thead>
                <tr>
                    <th scope="col">Partition</th>
                    <th scope="col">Latest Offset</th>
                    <th scope="col">Leader</th>
                    <th scope="col">Replicas</th>
                    <th scope="col">In Sync Replicas</th>
                    <th scope="col">Preferred Leader?</th>
                    <th scope="col">Under Replicated?</th>
                    @if(topicIdentity.clusterContext.config.jmxEnabled && topicIdentity.clusterContext.config.displaySizeEnabled){
                    <th scope="col">Leader Size</th>
                    }
                </tr>
                </thead>
                <tbody>
                @for((_,tpi) <- topicIdentity.partitionsIdentity.toIndexedSeq.sortBy(_._2.partNum)) {
                <tr>
                    <th scope="row">@tpi.partNum</th>
                    <td>@tpi.latestOffset.map(_.formattedAsDecimal).getOrElse(" ")</td>
                    <td class="@getPartitionLeaderLevel(tpi.leader)">
                        <a href="@routes.Cluster.broker(cluster,tpi.leader)">@tpi.leader</a>
                    </td>
                    <td style="word-wrap: break-word">@tpi.replicas.mkString("(", ",", ")")</td>
                    <td style="word-wrap: break-word">@tpi.isr.mkString("(", ",", ")")</td>
                    <td class="@getIsPreferredLeaderLevel(tpi.isPreferredLeader)">
                        @tpi.isPreferredLeader
                    </td>
                    <td class="@getIsUnderReplicatedLevel(tpi.isUnderReplicated)">
                        @tpi.isUnderReplicated
                    </td>
                    @if(topicIdentity.clusterContext.config.jmxEnabled && topicIdentity.clusterContext.config.displaySizeEnabled){
                    <td>@tpi.size</td>
                    }
                </tr>
                }
                </tbody>
            </table>
            </div>
        </div>
    </div>
</div>
